package com.sun.stringescapeutils;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class CreatFluxTest {
    public static void main(String[] args) {
        //方法1
        Flux<Integer> justFlux = Flux.just(1, 2, 3, 4, 5, 6);//变成流
        subscribeFlux("justFlux", justFlux);

        //方法2
        Flux<Integer> arrayFlux = Flux.fromArray(new Integer[]{1, 2, 3, 4, 5, 6});
        subscribeFlux("arrayFlux", arrayFlux);

        //方法3
        List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5, 6);
        Flux<Integer> iterableFlux = Flux.fromIterable(integers);
        subscribeFlux("iterableFlux", iterableFlux);

        //方法4
        Stream<Integer> integerStream = Stream.of(1, 2, 3, 4, 5, 6);
        Flux<Integer> streamFlux = Flux.fromStream(integerStream);
        subscribeFlux("streamFlux", streamFlux);

        //方法5
        Flux<Integer> rangeFlux = Flux.range(1, 6);
        subscribeFlux("rangeFlux", rangeFlux);

        //方法6
        createFluxProgrammatically();

        //互相转换
        Mono<List<Integer>> listMono = Flux.just(1, 2, 3).collectList();
        Flux<Integer> flux = Mono.just(2).flux();
    }


    private static void subscribeFlux(String varName, Flux<?> flux) {
        flux.doOnSubscribe(s -> System.out.print(varName + ":"))
                .doOnNext(e -> System.out.print(e + ", "))
                .doOnComplete(System.out::println)
                .subscribe();//没有这个，流中不会有东西
    }

    public static void createFluxProgrammatically() {
        Flux<Object> generate = Flux.generate(() -> 1, (state, sink) -> {
            sink.next("message #" + state);
            if (state == 10) {
                sink.complete();
            }
            return state + 1;
        });
        subscribeFlux("generate", generate);
    }
}
